package server

import (
	"bufio"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"github.com/pion/rtcp"
	"github.com/pion/rtp"
	"log"
	"net"
	"strings"
	"time"

	"gitee.com/baidugoogle/rtsp-server/protocol/base"
	"gitee.com/baidugoogle/rtsp-server/protocol/headers"
	"github.com/general252/gout/ucolor"
	"github.com/general252/gout/ustring"
)

const (
	networkBufferSize    = 512 * 1024
	serverReadBufferSize = 4096
)

type ConnectionType int

const (
	ConnectionTypeUnknown ConnectionType = 0
	ConnectionTypePusher  ConnectionType = 1
	ConnectionTypePlayer  ConnectionType = 2
)

type RtspSession struct {
	server *RtspServer

	sessionId string
	conn      *net.TCPConn

	connType      ConnectionType
	mStreamPusher *StreamPusher
	mStreamPlayer *StreamPlayer

	tcpTracksByChannel map[int]*PusherTrack // tcp 通过setup的interleaved匹配track

	baseURL *base.URL
}

func NewConnection(conn *net.TCPConn, s *RtspServer) *RtspSession {
	var buff [4]byte
	_, _ = rand.Read(buff[:])

	c := &RtspSession{
		server:    s,
		sessionId: hex.EncodeToString(buff[:]),
		conn:      conn,
		connType:  ConnectionTypeUnknown,

		tcpTracksByChannel: map[int]*PusherTrack{},
	}

	c.server.getDispatchHandler().OnSessionOpen(c)

	go c.run()

	return c
}

func (c *RtspSession) Close() {
	res := base.Request{
		Method: base.Teardown,
		URL:    c.baseURL,
		Header: make(base.Header),
		Body:   nil,
	}

	res.Header["CSeq"] = base.HeaderValue{"5"}
	res.Header["RtspServer"] = base.HeaderValue{"rtsp-RtspServer"}
	res.Header["Session"] = base.HeaderValue{c.sessionId}

	_, _ = c.conn.Write([]byte(res.String()))

	time.Sleep(time.Millisecond * 100)

	_ = c.conn.Close()
}

func (c *RtspSession) GetUid() string {
	return c.sessionId
}

// GetConnectionType 连接类型(推流、拉流)
func (c *RtspSession) GetConnectionType() ConnectionType {
	return c.connType
}

func (c *RtspSession) IsPlayer() bool {
	return c.connType == ConnectionTypePlayer
}

func (c *RtspSession) IsPusher() bool {
	return c.connType == ConnectionTypePusher
}

// GetStreamPusher 获取推流器
func (c *RtspSession) GetStreamPusher() *StreamPusher {
	return c.mStreamPusher
}

func (c *RtspSession) GetClientAddress() (*net.TCPAddr, bool) {
	if c == nil || c.conn == nil {
		return nil, false
	}
	addr, ok := c.conn.RemoteAddr().(*net.TCPAddr)
	if !ok || addr == nil {
		return nil, false
	}

	return addr, true
}

func (c *RtspSession) responseMessage(statusCode base.StatusCode, a ...interface{}) (*base.Response, error) {
	return &base.Response{
		StatusCode:    statusCode,
		StatusMessage: ustring.Format(a...),
		Header:        base.Header{},
	}, nil
}

func (c *RtspSession) getSessionID(header base.Header) string {
	if h, ok := header["Session"]; ok && len(h) == 1 {
		return h[0]
	}
	return ""
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

func (c *RtspSession) run() {

	const (
		tcpMaxFramePayloadSize = 3 * 1024 * 1024

		// 1500 (UDP MTU) - 20 (IP header) - 8 (UDP header)
		maxPacketSize = 1472
	)

	defer func() {
		log.Printf("session type %v", c.connType)

		if c.mStreamPlayer != nil {
			c.mStreamPlayer.Close()
		}
		if c.mStreamPusher != nil {
			c.mStreamPusher.Close()
		}

		_ = c.conn.Close()
		c.server.getDispatchHandler().OnSessionClose(c)

		log.Printf(">>>>> close session %v", c.sessionId)
	}()

	var handleMethod = func(req *base.Request) error {
		if c.baseURL == nil {
			c.baseURL = req.URL
		}

		log.Printf("%s\n%s", ucolor.HiGreen("=========== %v ===========", req.Method), req.String())

		// 处理请求
		res, err := c.handleRequest(req)
		if err != nil {
			return err
		}

		if res.Header == nil {
			res.Header = make(base.Header)
		}
		res.Header["CSeq"] = req.Header["CSeq"]
		res.Header["RtspServer"] = base.HeaderValue{"rtsp-RtspServer"}
		res.Header["Session"] = base.HeaderValue{c.getSessionID(req.Header)}

		log.Printf("%s\n%s", ucolor.Yellow("%v %s", base.StatusMessages[res.StatusCode], res.StatusMessage), res.String())

		// 发送回复
		if _, err = c.conn.Write([]byte(res.String())); err != nil {
			return err
		}

		return nil
	}

	var handleFrame = func(track *PusherTrack, frame *base.InterleavedFrame) error {
		isRTP := (frame.Channel % 2) == 0
		data := frame.Payload

		if isRTP {
			// rtp
			var pkt rtp.Packet
			if err := pkt.Unmarshal(data); err != nil {
				return err
			}

			// log.Printf("recv rtp %v", pkt.SSRC)
			if err := track.WriteRTP(&pkt, data); err != nil {
				return err
			}
		} else {
			// rtcp
			if len(data) > maxPacketSize {
				return fmt.Errorf("payload size (%d) greater than maximum allowed (%d)", len(data), maxPacketSize)
			}

			// 多个rtcp包
			rtcpPackets, err := rtcp.Unmarshal(data)
			if err != nil {
				return err
			}

			// 处理每个rtcp
			for _, packet := range rtcpPackets {
				switch rtcpPacket := packet.(type) {
				case *rtcp.SenderReport:
					// 发送者报告
					log.Printf("recv rtcp %v", rtcpPacket.SSRC)
					track.OnRtcpPacket(&RtcpPacket{
						Packet:         rtcpPacket,
						FromConnection: nil,
						FromAddr:       nil,
					})

					log.Printf("tcp sender report. %v(0x%X) info: %v", rtcpPacket.SSRC, rtcpPacket.SSRC, rtcpPacket.String())
				}
			}
		}

		return nil
	}

	var (
		req          base.Request
		frame        base.InterleavedFrame
		framePayload = make([]byte, tcpMaxFramePayloadSize)
		connReader   = bufio.NewReaderSize(c.conn, serverReadBufferSize)
	)

	for {
		frame.Payload = framePayload
		what, err := base.ReadInterleavedFrameOrRequest(&frame, &req, connReader)
		if err != nil {
			log.Println(err)
			break
		}

		switch what := what.(type) {
		case *base.Request:
			// rtsp method
			if err = handleMethod(what); err != nil {
				log.Println(err)
				return
			}
		case *base.InterleavedFrame:
			// rtp over tcp
			if track, ok := c.tcpTracksByChannel[what.Channel]; ok {
				if err = handleFrame(track, what); err != nil {
					log.Printf("handleFrame fail. %v", err)
				}
			} else {
				log.Printf("not found track. channel: %v, map: %v", what.Channel, c.tcpTracksByChannel)
			}
		}
	}

}

// handleRequest 处理请求
func (c *RtspSession) handleRequest(req *base.Request) (resp *base.Response, err error) {
	if cSeq, ok := req.Header["CSeq"]; !ok || len(cSeq) != 1 {
		return c.responseMessage(base.StatusBadRequest, "no CSeq")
	}

	if len(c.getSessionID(req.Header)) == 0 {
		req.Header["Session"] = base.HeaderValue{c.sessionId}
	}

	// 推流流程: Options > Announce > Setup > Setup > Record > Teardown

	// 拉流流程: Options > Describe > Setup > Setup > Play > Teardown

	switch req.Method {
	case base.Options:
		//
		resp, err = c.handleOptions(req)
		return
	case base.Announce:
		// 推流 req: sdp
		resp, err = c.handleAnnounce(req)
		c.server.getDispatchHandler().OnAnnounce(c)
		return
	case base.Describe:
		// 拉流 res: sdp
		resp, err = c.handleDescribe(req)
		c.server.getDispatchHandler().OnDescribe(c)
		return
	case base.Setup:
		// 协商端口号
		resp, err = c.handleSetup(req)
		return
	case base.Record:
		// 开始推流
		resp, err = c.handleRecord(req)
		c.server.getDispatchHandler().OnRecord(c)
		return
	case base.Play:
		// 开始拉流
		resp, err = c.handlePlay(req)
		c.server.getDispatchHandler().OnPlay(c)
		return
	case base.Pause:
		resp, err = c.handlePause(req)
		return
	case base.GetParameter:
		// 可以用于连接保活
		resp, err = c.handleGetParameter(req)
		return
	case base.SetParameter:
		//

	case base.Teardown:
		// 关闭
		resp, err = c.handleTeardown(req)
		return
	}

	return c.responseMessage(base.StatusNotImplemented)
}

// options
func (c *RtspSession) handleOptions(req *base.Request) (*base.Response, error) {
	_ = req

	var methods []base.Method
	methods = append(methods, base.Announce)
	methods = append(methods, base.Describe)
	methods = append(methods, base.GetParameter)
	methods = append(methods, base.Options)
	methods = append(methods, base.Pause)
	methods = append(methods, base.Play)
	methods = append(methods, base.Record)
	methods = append(methods, base.Setup)
	methods = append(methods, base.SetParameter)
	methods = append(methods, base.Teardown)

	var supportMethodList []string
	for _, method := range methods {
		supportMethodList = append(supportMethodList, string(method))
	}

	return &base.Response{
		StatusCode: base.StatusOK,
		Header: base.Header{
			"Public": base.HeaderValue{strings.Join(supportMethodList, ", ")},
		},
	}, nil
}

// announce 推流, sdp
func (c *RtspSession) handleAnnounce(req *base.Request) (*base.Response, error) {
	//  rtsp://127.0.0.1:554/test

	// pathAndQuery: test
	pathAndQuery, ok := req.URL.RTSPPathAndQuery()
	if !ok {
		return c.responseMessage(base.StatusBadRequest, "no path")
	}

	// path: test
	path, query := base.PathSplitQuery(pathAndQuery)

	ct, ok := req.Header["Content-Type"]
	if !ok || len(ct) != 1 {
		return c.responseMessage(base.StatusBadRequest, "Content-Type header is missing")
	}

	if ct[0] != "application/sdp" {
		return c.responseMessage(base.StatusBadRequest, "error sdp")
	}

	/*
		v=0
		o=- 0 0 IN IP4 127.0.0.1
		s=No Name
		c=IN IP4 127.0.0.1
		t=0 0
		a=tool:libavformat 58.76.100
		m=video 0 RTP/AVP 96
		a=rtpmap:96 H264/90000
		a=fmtp:96 packetization-mode=1; sprop-parameter-sets=Z/QAH5GbKBkCb8TCAAADABgAAAMDxB4wYyw=,aOvjxEhE; profile-level-id=F4001F
		a=control:streamid=0

		v=0
		o=- 0 0 IN IP4 127.0.0.1
		s=No Name
		c=IN IP4 127.0.0.1
		t=0 0
		a=tool:libavformat 58.76.100
		m=video 0 RTP/AVP 96
		a=rtpmap:96 H264/90000
		a=fmtp:96 packetization-mode=1; sprop-parameter-sets=Z2QAH6zZQFAFuhAAAD6QAA6mAPGDGWA=,aOvjyyLA; profile-level-id=64001F
		a=control:streamid=0
		a=ssrc:258369 cname:nYAodpnTebS8lziQ
		m=audio 0 RTP/AVP 97
		b=AS:128
		a=rtpmap:97 MPEG4-GENERIC/48000/2
		a=fmtp:97 profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3; config=119056E500
		a=ssrc:123458 cname:nYAodpnTebS8lziR
		a=control:streamid=1
	*/
	var sdp = req.Body

	// Media: video
	//
	// Attributes
	// rtpmap:96 H264/90000
	// fmtp:96 packetization-mode=1; sprop-parameter-sets=Z/QAH5GbKBkCb8TCAAADABgAAAMDxB4wYyw=,aOvjxEhE; profile-level-id=F4001F
	// control:stream=0
	tracks, err := ReadTracks(sdp)
	if err != nil {
		return c.responseMessage(base.StatusBadRequest, "invalid SDP. %v", err)
	}

	if len(tracks) == 0 {
		return c.responseMessage(base.StatusBadRequest, "no tracks defined in the SDP")
	}

	// 检查
	for _, track := range tracks {
		trackURL, err := track.URL(req.URL)
		if err != nil {
			return c.responseMessage(base.StatusBadRequest, "unable to generate track URL. %v", err)
		}

		// /test/streamid=0
		trackPath, ok := trackURL.RTSPPathAndQuery()
		if !ok {
			return c.responseMessage(base.StatusBadRequest, "invalid track URL (%v)", trackURL)
		}

		if !strings.HasPrefix(trackPath, path) {
			return c.responseMessage(base.StatusBadRequest, "invalid track path: must begin with '%s', but is '%s'", path, trackPath)
		}
	}

	c.connType = ConnectionTypePusher
	// 创建pusher
	c.mStreamPusher = NewStreamPusher(path, query, string(sdp), tracks, c)

	return c.responseMessage(base.StatusOK)
}

// setup
func (c *RtspSession) handleSetup(req *base.Request) (*base.Response, error) {
	/*
		SETUP rtsp://127.0.0.1:554/test/streamid=0 RTSP/1.0
		CSeq: 3
		Session: 05a1af64
		Transport: RTP/AVP/UDP;unicast;client_port=27202-27203;mode=record
		User-Agent: Lavf58.76.100


		RTP/AVP/UDP;unicast;server_port=8052;client_port=14186;ssrc=0B6020AD;mode=PLAY
	*/

	/*
		SETUP rtsp://127.0.0.1:554/test/streamid=0 RTSP/1.0
		CSeq: 3
		Session: daa5341c
		Transport: RTP/AVP/TCP;unicast;interleaved=0-1;mode=record
		User-Agent: Lavf58.76.100

		SETUP rtsp://127.0.0.1:554/test/streamid=1 RTSP/1.0
		CSeq: 4
		Session: daa5341c
		Transport: RTP/AVP/TCP;unicast;interleaved=2-3;mode=record
		User-Agent: Lavf58.76.100
	*/

	var inTransport headers.Transport
	if err := inTransport.Read(req.Header["Transport"]); err != nil {
		return c.responseMessage(base.StatusBadRequest, "invalid transport header. %v", err)
	}

	// 检查参数
	{
		switch c.connType {
		case ConnectionTypePlayer:
			if c.mStreamPlayer == nil {
				return c.responseMessage(base.StatusBadRequest, "no player object")
			}
		case ConnectionTypePusher:
			if c.mStreamPusher == nil {
				return c.responseMessage(base.StatusBadRequest, "no pusher object")
			}
		default:
			return c.responseMessage(base.StatusBadRequest, "unknown connection type. %v", c.connType)
		}

		//  udp 必须有ClientPorts属性
		if inTransport.Protocol == headers.TransportProtocolUDP && inTransport.ClientPorts == nil {
			return c.responseMessage(base.StatusBadRequest, "no client_port")
		}
	}

	var (
		trackControl string
		playTrack    *PlayerTrack
		pushTrack    *PusherTrack
	)

	// 提取control
	{
		// /test/streamid=0
		pathAndQuery, ok := req.URL.RTSPPathAndQuery()
		if !ok {
			return c.responseMessage(base.StatusBadRequest, "no path")
		}

		// /test/streamid=0
		pathAndControl, _ := base.PathSplitQuery(pathAndQuery)

		switch c.connType {
		case ConnectionTypePlayer:
			trackControl = strings.ReplaceAll(pathAndControl, c.mStreamPlayer.GetPath()+"/", "")
			if playTrack, ok = c.mStreamPlayer.GetTrackFromControl(trackControl); !ok {
				return c.responseMessage(base.StatusNotFound, "no found play track")
			}
		case ConnectionTypePusher:
			trackControl = strings.ReplaceAll(pathAndControl, c.mStreamPusher.GetPath()+"/", "")
			if pushTrack, ok = c.mStreamPusher.GetTrackFromControl(trackControl); !ok {
				return c.responseMessage(base.StatusNotFound, "no found pusher track")
			}

			if inTransport.SSRC != nil {
				// setup时传递ssrc
				pushTrack.SetSSRC(*inTransport.SSRC)
			}
		}

		// streamid=0
		log.Printf("setup %v", trackControl)
	}

	de := headers.TransportDeliveryUnicast
	var outTransport *headers.Transport

	// 拉流
	if c.connType == ConnectionTypePlayer {
		// mode := headers.TransportModePlay
		ssrc := playTrack.GetSSRC()

		outTransport = &headers.Transport{
			Protocol:       inTransport.Protocol,
			Delivery:       &de,
			InterleavedIDs: inTransport.InterleavedIDs,
			ClientPorts:    inTransport.ClientPorts,
			ServerPorts:    nil,
			Mode:           inTransport.Mode,
			SSRC:           &ssrc, // 发送给客户端ssrc
		}

		playTrack.inTransport = &inTransport
		if inTransport.Protocol == headers.TransportProtocolUDP {
			outTransport.ServerPorts = &[2]int{c.server.GetRtpPort(), c.server.GetRtcpPort()}
			playTrack.rtpClientPort = inTransport.ClientPorts[0]  // 记录客户端rtp端口
			playTrack.rtcpClientPort = inTransport.ClientPorts[1] // 记录客户端rtcp端口

			log.Printf("[%v] ssrc: %v, rtp: %v, rtcp: %v control: %v",
				ucolor.HiRed("player"),
				ssrc, playTrack.rtpClientPort, playTrack.rtcpClientPort, trackControl)
		} else if inTransport.Protocol == headers.TransportProtocolTCP {
			// nothing
		}
	}

	// 推流
	if c.connType == ConnectionTypePusher {
		outTransport = &headers.Transport{
			Protocol:       inTransport.Protocol,
			Delivery:       &de,
			InterleavedIDs: inTransport.InterleavedIDs,
			ClientPorts:    inTransport.ClientPorts,
			ServerPorts:    nil,
			Mode:           inTransport.Mode,
		}

		pushTrack.inTransport = &inTransport
		if inTransport.Protocol == headers.TransportProtocolUDP {
			outTransport.ServerPorts = &[2]int{c.server.GetRtpPort(), c.server.GetRtcpPort()}
			pushTrack.rtpClientPort = inTransport.ClientPorts[0]  // 记录客户端rtp端口, 没啥用
			pushTrack.rtcpClientPort = inTransport.ClientPorts[1] // 记录客户端rtcp端口, 发送给客户端"接收者报告"

			log.Printf("[%v] rtp: %v, rtcp: %v control: %v",
				ucolor.HiRed("pusher"),
				pushTrack.rtpClientPort, pushTrack.rtcpClientPort, trackControl)
		} else if inTransport.Protocol == headers.TransportProtocolTCP {
			// 记录over tcp中chanel和track的对应关系, tcp接收到rtp数据后, 通过chanel获取对应track
			c.tcpTracksByChannel[inTransport.InterleavedIDs[0]] = pushTrack // rtp over tcp, 记录channel对应track
			c.tcpTracksByChannel[inTransport.InterleavedIDs[1]] = pushTrack // rtp over tcp, 记录channel对应track

			log.Printf("[%v] channel: %v, %v control: %v",
				ucolor.HiRed("pusher"),
				inTransport.InterleavedIDs[0], inTransport.InterleavedIDs[1], trackControl)
		}
	}

	// 回复
	return &base.Response{
		StatusCode: base.StatusOK,
		Header: base.Header{
			"Transport": outTransport.Write(),
		},
	}, nil
}

// record
func (c *RtspSession) handleRecord(req *base.Request) (*base.Response, error) {
	return &base.Response{
		StatusCode: base.StatusOK,
	}, nil
}

// describe
func (c *RtspSession) handleDescribe(req *base.Request) (*base.Response, error) {
	var err error

	pathAndQuery, ok := req.URL.RTSPPathAndQuery()
	if !ok {
		return c.responseMessage(base.StatusBadRequest, "no path")
	}

	path, query := base.PathSplitQuery(pathAndQuery)

	pusher, found := c.server.GetPusher(path)
	if !found {
		return c.responseMessage(base.StatusNotFound, "not found path")
	}

	c.connType = ConnectionTypePlayer
	if c.mStreamPlayer, err = NewStreamPlayer(c, pusher, query); err != nil {
		return c.responseMessage(base.StatusInternalServerError, "new player fail. %v", err)
	}

	res := &base.Response{
		StatusCode: base.StatusOK,
		Header: base.Header{
			"Content-Base": base.HeaderValue{req.URL.String() + "/"},
			"Content-Type": base.HeaderValue{"application/sdp"},
		},
		Body: c.mStreamPlayer.tracks.Write(false),
	}

	return res, nil
}

// play
func (c *RtspSession) handlePlay(req *base.Request) (*base.Response, error) {
	c.mStreamPlayer.Play()

	return &base.Response{
		StatusCode: base.StatusOK,
	}, nil
}

// pause
func (c *RtspSession) handlePause(req *base.Request) (*base.Response, error) {
	if c.connType == ConnectionTypePlayer {
		c.mStreamPlayer.Pause()
	}

	return &base.Response{
		StatusCode: base.StatusOK,
	}, nil
}

// getParameter
func (c *RtspSession) handleGetParameter(req *base.Request) (*base.Response, error) {
	return &base.Response{
		StatusCode: base.StatusOK,
	}, nil
}

// teardown
func (c *RtspSession) handleTeardown(req *base.Request) (*base.Response, error) {

	var closeSession = func() {
		log.Printf("teardown session %v %v", c.sessionId, c.connType)
		if c.connType == ConnectionTypePlayer {
			c.mStreamPlayer.Close()
		} else {
			c.mStreamPusher.Close()
		}
	}

	go func() {
		time.Sleep(time.Second)

		closeSession()
	}()

	return &base.Response{
		StatusCode: base.StatusOK,
	}, nil
}

func (c *RtspSession) readRequest(br *bufio.Reader) (*base.Request, error) {
	var req base.Request
	if err := req.Read(br); err != nil {
		return nil, err
	}

	return &req, nil
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
